package org.kunteng.yunac.util;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ZakerRecord {
    static String es_url = "http://192.168.20.25:9200/wifimh/zaker_record/_search";
    
    public static void main(String[] args) {
	List<Map<String, String>> list = queryUVPVFromES();
	if (list != null && list.size() > 0) {
	    savePVUV(list);
	}
	
	saveContent("2016-08-27");
	saveContent("2016-08-28");
	
	String sql = "UPDATE zaker_content c,zaker_record r SET c.pv=r.pv,c.uv=r.uv WHERE c.contentId=r.id";
	executeSql(sql);
	
	statistic("2016-08-27");
	statistic("2016-08-28");
    }

    public static List<Map<String, String>> queryUVPVFromES() {
	ElasticSearchClient client = ElasticSearchClient.getInstance();
	String jsonString = "{\"_source\":[\"contentId\"],\"query\":{\"bool\":{\"must\":[{\"range\":{\"@timestamp\":{\"gte\":\"${{startTime}}\",\"lte\":\"${{endTime}}\"}}}]}},\"aggs\":{\"group_by\":{\"terms\":{\"field\":\"contentId\",\"size\":\"0\"},\"aggs\":{\"distinct\":{\"cardinality\":{\"field\":\"cookieId\"}}}}}}";

	/* 由于ElasticSearch时区晚了8小时，查询区间开始时间和结束时间都需要减8小时 */
//	SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
//	Calendar cal = Calendar.getInstance();
//	cal.setTime(new Date());
//	cal.add(Calendar.DATE, -2);
//	String startDate = sdf.format(cal.getTime());
//	String startTime = startDate + "T16:00:00";
//	cal.add(Calendar.DATE, 1);
//	String endDate = sdf.format(cal.getTime());
//	String endTime = endDate + "T15:59:59";
	String startTime = "1472227200000";
	String endTime="1472399999000";
	
	jsonString = jsonString.replace("${{startTime}}", startTime).replace("${{endTime}}", endTime);
	Map response = client.post(es_url, jsonString);
	return resolveGroupDistinctES(response);
    }

    /**
     * 解析ES接口返回的分组并去重数据
     * 
     * @param response
     * @return
     */
    public static List<Map<String, String>> resolveGroupDistinctES(Map response) {
	List<Map<String, String>> list = new ArrayList<Map<String, String>>();
	
	try {
	    List rs = (List) ((Map) ((Map) response.get("aggregations")).get("group_by")).get("buckets");
	    for (Object item : rs) {
		try {
		    Map jsItem = (Map) item;
		    String key = (String) jsItem.get("key");
		    int pv = ((Integer) jsItem.get("doc_count")).intValue();
		    Map distinct = (Map) jsItem.get("distinct");
		    int uv = ((Integer) distinct.get("value")).intValue();
		    HashMap<String, String> hm = new HashMap<String, String>();
		    hm.put("id", key);
		    hm.put("pv", String.valueOf(pv));
		    hm.put("uv", String.valueOf(uv));
		    list.add(hm);
		} catch (Exception e) {
		    e.printStackTrace();
		}
	    }
	} catch (Exception e) {
	    e.printStackTrace();
	}
	return list;
    }
    
    public static int executeSql(String sql) {
	int flag = 0;
	Connection conn = new DBConn().getConn();
	try {
	    Statement st = conn.createStatement();
	    st.execute(sql);
	    st.close();
	    conn.close();

	    flag = 1;
	} catch (Exception e) {
	    e.printStackTrace();
	}
	return flag;
    }
    
    public static int savePVUV(List<Map<String, String>> list) {
	int flag = 0;

	Connection conn = new DBConn().getConn();
	try {
	    conn.setAutoCommit(false);
	    Statement st = conn.createStatement();
	    for (Map<String, String> map : list) {
		String sql = "INSERT INTO zaker_record(id, pv, uv) VALUES('"+ map.get("id") + "'," + map.get("pv") + ","+ map.get("uv") + ")";
		st.addBatch(sql);
	    }
	    st.executeBatch();
	    conn.commit();
	    st.close();
	    conn.close();

	    flag = 1;
	} catch (Exception e) {
	    e.printStackTrace();
	    try {
		conn.rollback();
	    } catch (SQLException e1) {
		e1.printStackTrace();
	    }
	}

	return flag;
    }
    
    public static int saveContent(String date) {
	int flag = 0;

	Connection conn = new DBConn().getConn();
	try {
	    conn.setAutoCommit(false);
	    Statement st = conn.createStatement();
	    for (int n=1;n<=8;n++) {
		for (int i=0;i<24;i++) {
			String hour = String.valueOf(i);
			if (i<10) {
			    hour = "0" + String.valueOf(i);
			}
			String sql = "INSERT INTO zaker_content ";
			sql += "SELECT REPLACE(UUID(),'-',''),@rownum:=@rownum+1 AS rownum,create_time,id,channel_id,0,0 ";
			sql += "FROM (SELECT @rownum:=0) r, content_settings ";
			sql += "WHERE DATE_FORMAT(create_time,'%Y-%m-%d %H:%i:%s') BETWEEN '"+date+" "+hour+":00:00' AND '"+date+" "+hour+":59:59' ";
			sql += "AND channel_id='"+n+"' ORDER BY create_time ASC";
			st.addBatch(sql);
		    }
	    }
	    st.executeBatch();
	    conn.commit();
	    st.close();
	    conn.close();

	    flag = 1;
	} catch (Exception e) {
	    e.printStackTrace();
	    try {
		conn.rollback();
	    } catch (SQLException e1) {
		e1.printStackTrace();
	    }
	}

	return flag;
    }
    
    public static int statistic(String date,String hour,int cid) {
	int flag = 0;
	
	try {
	    int count = 0;
	    Connection conn = new DBConn().getConn();
	    Statement st = conn.createStatement();
	    String sql = "SELECT count(1) AS count FROM zaker_content ";
	    sql += "WHERE DATE_FORMAT(createTime,'%Y-%m-%d %H:%i:%s') BETWEEN '"+date+" "+hour+":00:00' AND '"+date+" "+hour+":59:59' ";
	    sql += "AND channelId='"+cid+"'  ORDER BY rownum ASC";
	    ResultSet re = st.executeQuery(sql);
	    while (re.next()) {
		count = re.getInt("count");
	    }
	    
//	    System.out.println(date+" "+hour+" "+cid+" "+"count="+count);
	    if (count >0 ) {
		List<String> list = new ArrayList<String>();
		int num =1;
		for (int i=1;i<=count;i++) {
		    if (i == num) {
			list.add(String.valueOf(num));
			num += 5;
		    }
		}
		System.out.println(list.size());
		
		Connection conn1 = new DBConn().getConn();
		conn1.setAutoCommit(false);
		Statement st1 = conn1.createStatement();
		    String sql1 = "SELECT * FROM zaker_content ";
		    sql1 += "WHERE DATE_FORMAT(createTime,'%Y-%m-%d %H:%i:%s') BETWEEN '"+date+" "+hour+":00:00' AND '"+date+" "+hour+":59:59' ";
		    sql1 += "AND channelId='"+cid+"'  ORDER BY rownum ASC";
		    ResultSet re1 = st1.executeQuery(sql1);
		    while (re1.next()) {
			int rownum = re1.getInt("rownum");
			
			if (list.contains(String.valueOf(rownum))) {
			    System.out.println("rownum="+rownum);
			    String contentId = re1.getString("contentId");
			    String channelId = re1.getString("channelId");
			    int pv = re1.getInt("pv");
			    int uv = re1.getInt("uv");
			    
			    String insert_sql = "INSERT INTO zaker_statistic VALUES('"+date+"','"+hour+"','"+channelId+"','"+contentId+"',"+rownum+","+pv+","+uv+")";
			     st1.addBatch(insert_sql);
			    
			}
			
		    }
		     st1.executeBatch();
		     conn1.commit();
		    st1.close();
		    conn1.close();
	    }
	    
	    st.close();
	    conn.close();

	    flag = 1;
	} catch (Exception e) {
	    e.printStackTrace();
	}
	return flag;
    }
    
    public static void statistic(String date) {
	for (int n = 1; n <= 8; n++) {
	    for (int i = 0; i < 24; i++) {
		String hour = String.valueOf(i);
		if (i < 10) {
		    hour = "0" + String.valueOf(i);
		}
		statistic(date,hour,n);
	    }
	}
    }
    

}
